package com.ods;

import com.utils.KafkaUtils;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

/**
 * 实现log日志发送到kafka  ods_base_log
 */
public class OdsLog {
    public static void main(String[] args) throws Exception {
        //流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //获取消费者  消费从kafka上发送的log日志入局
        FlinkKafkaConsumer<String> practice = KafkaUtils.createConsumer("practice","dsawrv");
        DataStreamSource<String> ods_LogDs = env.addSource(practice);
        //将数据发送到dwd层进行基础的分层
        FlinkKafkaProducer ods_base_log = KafkaUtils.createProduer("ods_base_log");
        ods_LogDs.addSink(ods_base_log);

        env.execute();

    }
}
